package tang.client;

import lombok.Data;
import tang.common.ConnectResponse;
import tang.common.CreateResponse;
import tang.common.OpCode;
import tang.common.Packet;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.LinkedList;

/**
 * @author Tang
 * @classname ClientCnxn
 * @description [  ]
 * @date 2021/2/23 21:02
 */
@Data
public class ClientCnxn {

    private SendThread sendThread;

    /**
     * 连接状态
     */
    private Zookeeper.States state = Zookeeper.States.NOT_CONNECTED;

    private ClientCnxnSocketNIO clientCnxnSocketNIO;

    private InetSocketAddress address;

    /**
     * 需要发送的队列
     */
    private final LinkedList<Packet> outgoingQueue = new LinkedList<>();

    /**
     * 待处理响应的队列
     */
    private final LinkedList<Packet> pendingQueue = new LinkedList<>();

    public ClientCnxn(InetSocketAddress address, ClientCnxnSocketNIO clientCnxnSocket) {
        sendThread = new SendThread();
        this.clientCnxnSocketNIO = clientCnxnSocket;
        this.clientCnxnSocketNIO.clientCnxn = this;
        this.clientCnxnSocketNIO.sendThread = sendThread;
        this.address = address;
    }

    public void start() {
        sendThread.start();
    }

    public void submitQueue(Packet packet) {
        outgoingQueue.add(packet);
        synchronized (packet) {
            while (!packet.finished) {
                try {
                    packet.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    class SendThread extends Thread {

        public SendThread() {
            super("zk-senThread");
            setDaemon(true);
        }

        @Override
        public void run() {
            // 活着
            while (state.isAlive()) {
                // 第一次连接
                if (!clientCnxnSocketNIO.isConnected()) {
                    startConnection(address);
                }
                // 连接成功
                if (state.isConnected()) {
//                    间隔指定时间，就发送心跳
//                    sendPing();
                    // 处理读写请求
                    try {
                        clientCnxnSocketNIO.doTransport(pendingQueue,outgoingQueue,ClientCnxn.this);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

        private void sendPing() {
            System.out.println("隔30s发送心跳包开始...");
        }

        private void startConnection(InetSocketAddress address) {
            state = Zookeeper.States.CONNECTING;
            try {
                clientCnxnSocketNIO.connect(address);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        public void readResponse(ByteBuffer incomingBuffer) {
            if (!pendingQueue.isEmpty()) {
                Packet packet = pendingQueue.removeFirst();
                // 根据操作类型 解析数据 incomingBuffer 填充response
                synchronized (packet) {
                    // 读模式
                    incomingBuffer.flip();
                    switch (packet.getType()) {
                        case Connect:
                            if (packet.response instanceof ConnectResponse) {
                                ConnectResponse connectResponse = (ConnectResponse) packet.response;
                                connectResponse.decoder(incomingBuffer);
                            }
                            break;
                        case Create:
                            if (packet.response instanceof CreateResponse) {
                                CreateResponse createResponse = (CreateResponse) packet.response;
                                createResponse.decoder(incomingBuffer);
                            }
                            break;
                        default:
                            break;
                    }
                    packet.finished = true;
                    packet.notify();
                }
            }
        }
    }
}